package com.zhaofujun.filequeue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;


public class FileQueueWithReactor<T> implements FileQueue<T> {

    Logger logger = LoggerFactory.getLogger(FileQueueWithReactor.class);

    private final StatusFileMapper globalContext;
    private final Serializer<T> serializer;
    private final SegmentFileMapper segmentContext;
    private final DeleteSegmentStrategy deleteSegmentStrategy;
    private final Runnable onClose;


    FileQueueWithReactor(String baseDir, int segmentSize, DeleteSegmentStrategy deleteSegmentStrategy, Serializer<T> serializer, Runnable onClose) throws IOException {

        this.segmentContext = new SegmentFileMapper(baseDir, segmentSize);
        this.globalContext = new StatusFileMapper(baseDir, segmentSize);
        this.deleteSegmentStrategy = deleteSegmentStrategy;
        this.deleteSegmentStrategy.setFileQueue(this);
        this.deleteSegmentStrategy.start();

        this.serializer = serializer;
        this.onClose = onClose;



    }


    @Override
    public void push(T t) throws IOException {

        byte[] bytes = serializer.serialize(t);
        DataInfo dataInfo = globalContext.preallocate(bytes, segmentContext);
        segmentContext.put(dataInfo);
        globalContext.updateWritePosition();
    }

    @Override
    public T pop(int no) throws IOException {
        // 从全局上下文获取当前读指针
        long readNum = globalContext.getReadNums().get(no);
        long writeNum = globalContext.getWriteNumRef().getNum();
        if (readNum >= writeNum) {
            return null;
        }

        // 更新读指针
        if (globalContext.getReadNums().compareAndSet(no, readNum)) {
            // 检查当前位置是否有效，有效则读取，无效则递归调用

            T t = get(readNum);
            //更新文件中的位置信息
            globalContext.updateReadPosition(no);
            return t;
        } else {
            return pop(no);
        }
    }

    @Override
    public void clear() throws IOException {
        this.close();
        segmentContext.deleteAllFiles();

    }

    @Override
    public long size() throws IOException {
        return globalContext.getWriteNumRef().getNumRef().getNum() + 1;
    }

    @Override
    public long unReadMessages(int no) throws IOException {
        long writeNum = globalContext.getWriteNumRef().getNumRef().getNum();
        long readNum = globalContext.getReadNums().get(no);
        return writeNum - readNum;
    }

    @Override
    public T get(long position) throws IOException {
        SegmentInfo segmentInfo = segmentContext.calculateSegmentInfo(position);
        byte[] bytes = segmentContext.get(segmentInfo);
        return serializer.deserialize(bytes);
    }

    @Override
    public T getTop(int no) throws IOException {
        // 从全局上下文获取当前读指针
        long readPosition = globalContext.getReadNums().get(no);
        // 从文件中读取数据
        T t = get(readPosition);

        return t;
    }

    @Override
    public void deleteSegmentFile() {
        //根据读写指针，找到已经完成读取的片段文件，将其删除
        long[] readPositions = globalContext.getReadNums().getNums();
        // 找到除0以外的最小值
        Arrays.stream(readPositions).filter(num -> num != 0).min().ifPresent(min -> {
            // 删除所有小于min的片段文件
            //读取最小位置的片段文件，将其删除

            SegmentInfo segmentInfo = segmentContext.calculateSegmentInfo(min);
            long readSegmentIndex = segmentInfo.getSegmentNumber();
            for (long i = 0; i < readSegmentIndex; i++) {
                try {
                    segmentContext.deleteSegmentFile(i);
                } catch (IOException e) {
                    logger.warn("delete segment file failed:{}", e);
                }
            }
        });


    }

    @Override
    public void close() throws IOException {

        //关闭删除策略
        deleteSegmentStrategy.stop();

        try {
            segmentContext.release();
            globalContext.release();
        } catch (InvocationTargetException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }

        deleteSegmentFile();
        // 执行关闭回调
        if (onClose != null) {
            onClose.run();
        }
    }
}
